package com.dk.logc.business;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.dk.logc.config.CountFactory;
import com.dk.logc.domain.CountRole;
import com.dk.logc.domain.CountRoleSum;
import com.dk.logc.domain.FlogOnline;
import com.dk.logc.kafka.base.BaseRoleFlog;
import com.dk.logc.kafka.base.Event;
import com.dk.logc.mapper.CountRoleMapper;
import com.dk.logc.service.CountRoleService;
import com.dk.logc.service.CountRoleSumService;
import com.dk.logc.utils.DateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;

import static com.dk.logc.config.GlobalConfig.typeCategoryMap;

/**
 * @Author 林道敬
 * @Date 2024-01-05
 * @Description
 **/

@Component
@Slf4j
public class CountRoleBusiness {

    @Autowired
    private CountRoleMapper countRoleMapper;
    @Autowired
    private CountRoleSumService countRoleSumService;
    @Autowired
    private CountRoleService countRoleService;
    @Autowired
    private FlogEntryBusiness flogEntryBusiness;
    @Autowired
    private FlogOnlineBusiness flogOnlineBusiness;
    @Autowired
    private FlogPaymentBusiness flogPaymentBusiness;
    @Autowired
    private FlogLoginBusiness flogLoginBusiness;

    @KafkaListener(topics = {"dabaigame"},groupId = "game-kafka-cluster",containerFactory="batchFactory")
    public void receiver(List<ConsumerRecord<?, ?>> record, Acknowledgment ack){
        try{
            if(record == null){
                return;
            }
            Event event = new Event();
            byte[] obj=(byte[]) record.get(0).value();
            if(obj!=null) {
                String msg=new String((byte[]) obj,"UTF-8");
                event= JSONObject.parseObject(msg, Event.class);
            }
            byte[] body = event.getBody();
            Map eventMap  = event.getHeader();
            String channelId = (String) eventMap.get("channelId");
            String serverNo = (String) eventMap.get("serverNo");
            //用户数据
            String realValue = new String(body);
            CountRoleSum countRoleSum = initCountRoleSum(channelId, Integer.parseInt(serverNo));
            CountRole newCountRole = initCountRole(channelId, Integer.parseInt(serverNo),new Date());
            //类型
            String entityKey = (String) eventMap.get("category");
            //用户类
            Class realClass = typeCategoryMap.get(entityKey);
            BaseRoleFlog baseRoleFlog = (BaseRoleFlog) JSON.parseObject(realValue,realClass);
            baseRoleFlog.setCreateTime(new Date());
            baseRoleFlog.setServerNo(Integer.parseInt(serverNo));
            baseRoleFlog.setChannelId(channelId);
            CountFactory.get(entityKey).progress(baseRoleFlog,newCountRole,countRoleSum);
            ack.acknowledge();
        }catch (Exception e){
            log.error(e.getMessage());
        }
    }

    public CountRoleSum initCountRoleSum(String channelId, Integer serverNo) {
        CountRoleSum countRoleSum = new CountRoleSum();
        try{
            QueryWrapper<CountRoleSum> sumQueryWrapper = new QueryWrapper<>();
            sumQueryWrapper.eq("channel_id",channelId)
                    .eq("server_no",serverNo);
            countRoleSum = countRoleSumService.getOne(sumQueryWrapper);
            QueryWrapper<CountRole> countRoleQueryWrapper = new QueryWrapper<>();
            countRoleQueryWrapper.eq("channel_id",channelId)
                    .eq("server_no",serverNo);
            CountRole countRole = countRoleService.getOne(countRoleQueryWrapper);
            if(countRole == null){
                countRoleSum.setTotalRegister(0);
                countRoleSum.setChannelId(channelId);
                countRoleSum.setTotalRegister(0);
                countRoleSum.setServerNo(serverNo);
                countRoleSum.setTotalNewPayAmount(0);
                countRoleSum.setTotalNewPayNum(0);
                countRoleSum.setTotalOldPayAmount(0);
                countRoleSum.setTotalOldPayNum(0);
                return countRoleSum;
            }
            if(countRoleSum == null) {
                countRoleSum = countRoleMapper.queryAllByNum(channelId,serverNo);
                countRoleSum.setChannelId(channelId);
                countRoleSum.setServerNo(serverNo);
            }
        }catch (Exception e){
            log.error(e.getMessage());
        }
        return countRoleSum;
    }

    public CountRole initCountRole(String channelId, Integer serverNo,Date ymd) {
        Date start = DateUtil.strToDateLong(DateUtil.dateToStr(new Date(), Locale.CHINA) );

        QueryWrapper<CountRole> countRoleQueryWrapper  = new QueryWrapper<>();
        countRoleQueryWrapper.eq("channel_id",channelId)
                .eq("server_no",serverNo)
                .apply("to_days(ymd)  = to_days({0})",ymd);
        CountRole countRole = countRoleService.getOne(countRoleQueryWrapper);

        if(countRole == null){
            countRole = new CountRole(new Date(),channelId,serverNo,
                    0,0,0,0,0,0,0,0,0,0,0l,0l,0,0,0,0,0,0,0,0,0);
        }
        return countRole;
    }
}
